#include "metanet_server.h"

#include <stdbool.h>
#include <stdio.h>
#include <assert.h>
#include <string.h>

#include <pthread.h>

#include "metanet_imp.h"
#include "metanet_malloc.h"
#include "metanet_mq.h"
#include "metanet_module.h"
#include "metanet_handle.h"
#include "metanet.h"


struct metanet_context {
    void *inst;
    struct metanet_module *mod;
    uint32_t handle;
    int session_id;
    bool init;
    bool priority;
    int ref; //ying yong count

    void *cb_ud; //
    metanet_cb cb;
    struct metanet_message_queue *mq;
};


struct drop_t {
    uint32_t handle;
};

struct metanet_node {
    int total;
    int init;
    pthread_key_t handle_key;
};

static struct metanet_node G_NODE;

static void 
context_inc() {
    ++G_NODE.total;
}

static void
context_dec() {
    --G_NODE.total;
}

// static void
// id_to_hex(char *str, uint32_t id) {
//     int i;
//     static char hex[16] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'};
//     str[0] = ':';
//     for (i = 0; i < 8; ++i) {
//         str[i+1] = hex[(id>>((7-i) * 4) & 0xF)];
//     }
//     str[9] = '\0';
// }

static void
drop_message(struct metanet_message *msg, void *ud) {
    struct drop_t *d = ud;
    metanet_free(msg->data);
    uint32_t source = d->handle;
    assert(source);
    metanet_send(NULL, source, msg->source, PTYPE_ERROR, 0, NULL, 0);
}

struct metanet_context *
metanet_context_new(const char *name, const char *param) {
    struct metanet_module *mod = metanet_module_query(name);
    if (mod == NULL) {
        return NULL;
    }
    void *inst = metanet_module_instance_create(mod);
    if (inst == NULL) {
        return NULL;
    }
    struct metanet_context *ctx = metanet_malloc(sizeof(*ctx));

    ctx->mod = mod;
    ctx->inst = inst;
    ctx->ref = 2;
    ctx->cb = NULL;
    ctx->cb_ud = NULL;
    ctx->session_id = 0;

    ctx->init = false;
    ctx->priority = false;
    ctx->handle = 0;
    ctx->handle = metanet_handle_register(ctx);
    
    ctx->mq = metanet_mq_create(ctx->handle);
    struct metanet_message_queue *mq = ctx->mq;
    context_inc();

    int r = metanet_module_instance_init(mod, inst, ctx, param);
    if (r == 0) {
        struct metanet_context *ret = metanet_context_release(ctx);
        if (ret) {
            ctx->init = true;
        }
        metanet_globalmq_push(mq);
        if (ret) {
            metanet_error(ret, "LANCH %s %s", name, param ? param : "");
        }
        return ret;
    } else {
        metanet_error(ctx, "FAILED launch %s", name);
        uint32_t handle = ctx->handle;
        metanet_context_release(ctx);
        metanet_handle_retire(handle);
        struct drop_t d = {handle};
        metanet_mq_release(mq, drop_message, &d);
        return NULL;
    }
}

void 
metanet_context_grab(struct metanet_context *ctx) {
    ++ctx->ref;
}

void 
metanet_context_reserve(struct metanet_context *ctx) {
    metanet_context_grab(ctx);
    context_dec();
}

static void
delete_context(struct metanet_context *ctx) {
    metanet_module_instance_release(ctx->mod, ctx->inst);
    metanet_mq_mark_release(ctx->mq);
    metanet_free(ctx);
    context_dec();
}

struct metanet_context *
metanet_context_release(struct metanet_context *ctx) {
    if (ctx->ref == 0) {
        delete_context(ctx);
        return NULL;
    }
    return ctx;
}

uint32_t 
metanet_context_handle(struct metanet_context *ctx) {
   return ctx->handle;
}

int 
metanet_context_push(uint32_t handle, struct metanet_message *msg) {
    struct metanet_context *ctx = metanet_handle_grab(handle);
    if (ctx == NULL) {
        return -1;
    }
    metanet_mq_push(ctx->mq, msg);
    metanet_context_release(ctx);
    return 0;
}

void
metanet_context_send(struct metanet_context *ctx, void *msg, size_t sz, uint32_t source, int type, int session) {
    struct metanet_message smsg;
    smsg.source = source;
    smsg.session = session;
    smsg.data = msg;
    smsg.sz = sz | (size_t)type << MESSAGE_TYPE_SHIFT;
    metanet_mq_push(ctx->mq, &smsg);
}

int 
metanet_context_newsession(struct metanet_context *ctx) {
    int session = ++ctx->session_id;
    if (session <= 0) {
        ctx->session_id = 1;
        return 1;
    }
    return session;
}

static void
dispatch_message(struct metanet_context *ctx, struct metanet_message *msg) {
    assert(ctx->init);
    int type = msg->sz >> MESSAGE_TYPE_SHIFT;
    size_t sz = msg->sz & MESSAGE_TYPE_MASK;
    if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
        metanet_free(msg->data);
    }
}

struct metanet_message_queue *
metanet_context_message_dispatch(struct metanet_message_queue *mq, int weight) {
    if (mq == NULL) {
        mq = metanet_globalmq_pop();
        if (mq == NULL) {
            return NULL;
        }
    }
    uint32_t handle = metanet_mq_handle(mq);
    struct metanet_context *ctx = metanet_handle_grab(handle);
    if (ctx == NULL) {
        struct drop_t d = {handle};
        metanet_mq_release(mq, drop_message, &d);
        return metanet_globalmq_pop();
    }

    uint64_t i;
    uint64_t n = 1;
    struct metanet_message msg;

    for (i = 0; i < n; ++i) {
        if (metanet_mq_pop(mq, &msg)) {
            metanet_context_release(ctx);
            return metanet_globalmq_pop();
        } else if (i == 0 && weight > 0) {
            n = metanet_mq_length(mq);
            n >>= weight;
        }

        int overloaded = metanet_mq_overload(mq);
        if (overloaded) {
            //metanet_error(ctx, "May overloaded, message queue length = %d", overloaded);
        }
        if (ctx->cb == NULL) {
            metanet_free(msg.data);
        } else {
            dispatch_message(ctx, &msg);
        }
    }

    assert(mq == ctx->mq);
    struct metanet_message_queue *nq = metanet_globalmq_pop();
    if (nq) {
        if (mq->priority) {
            metanet_globalmq_push_head(mq);
        } else {
            metanet_globalmq_push(mq);
        }
        mq = nq;
    }

    return mq;
}

void 
metanet_context_message_dispatchall(struct metanet_context *ctx) {
    struct metanet_message msg;
    struct metanet_message_queue *mq = ctx->mq;
    while (metanet_mq_pop(mq, &msg)) {
        dispatch_message(ctx, &msg);
    }
}

void 
metanet_globalinit(void) {
    G_NODE.total = 0;
    G_NODE.init = 1;
    if (pthread_key_create(&G_NODE.handle_key, NULL)) {
        fprintf(stderr, "pthread_key_create failed\n");
        exit(1);
    }
}

void 
metanet_globalexit(void) {
    pthread_key_delete(G_NODE.handle_key);
}

void 
metanet_error(struct metanet_context *ctx, const char *msg, ...) {

}

const char *
metanet_command(struct metanet_context *ctx, const char *cmd, const char *param) {
    return "";
}

uint32_t 
metanet_queryname(struct metanet_context *ctx, const char *name) {
    switch(name[0]) {
    case ':':
        return strtoul(name+1, NULL, 16);
    case '.':
        return metanet_handle_findname(name+1);
    }
    metanet_error(ctx, "Don't support query global name %s", name);
    return 0;
}

static uint32_t
tohandle(struct metanet_context *ctx, const char *param) {
    uint32_t handle = 0;
    if (param[0] == ':') {

    } else if (param[0] == '.') {

    } else {
        metanet_error(ctx, "Cant't convert %s to handle", param);
    }
    return handle;
}

uint32_t 
metanet_service_key(struct metanet_context *ctx, const char *param) {
    uint32_t handle = tohandle(ctx, param);
    if (handle) {
        return handle;
    }
    return 0;
}

static void
_filter_args(struct metanet_context *ctx, int type, int *session, void **msg, size_t *sz) {
    int needcopy = !(type & PTYPE_TAG_DONTCOPY);
    int allocsession = type & PTYPE_TAG_ALLOCSESSION;
    type &= 0xFF;
    if (allocsession) {
        assert(*session == 0);
        *session = metanet_context_newsession(ctx);
    }

    if (needcopy && *msg) {
        char *smsg = metanet_malloc(sizeof(*sz + 1));
        memcpy(smsg, *msg, *sz);
        msg[*sz] = '\0';
        *msg = smsg;
    }
    *sz |= (size_t)type << MESSAGE_TYPE_SHIFT;
}

int 
metanet_send(struct metanet_context *ctx, uint32_t source, uint32_t destination, int type,  int session,  void *msg,  size_t sz) {
    if ((sz & MESSAGE_TYPE_MASK) != sz) {
        metanet_error(ctx, "The Message to %x is too large", destination);
        if (type & PTYPE_TAG_DONTCOPY) {
            metanet_free(msg);
        }
        return -2;
    }

    _filter_args(ctx, type, &session, (void**)&msg, &sz);
    if (source == 0) {
        source = ctx->handle;
    }

    if (destination == 0) {
        if (msg) {
            metanet_error(ctx, "Destination address can't be 0");
            metanet_free(msg);
            return -1;
        }
        return session;
    }

    struct metanet_message smsg;
    smsg.source = source;
    smsg.session = session;
    smsg.data = msg;
    smsg.sz = sz;
    if (metanet_context_push(destination, &smsg)) {
        metanet_free(msg);
        return -1;
    }

    return session;
}

int 
metanet_sendname(struct metanet_context *ctx, uint32_t source, const char *destination, int type, int session, void *msg, size_t sz) {
    if (source == 0) {
        source = ctx->handle;
    }
    uint32_t des = 0;
    if (destination[0] == ':') {
        des = strtol(destination + 1, NULL, 16);
    } else if (destination[0] == '.') {
        des = metanet_handle_findname(destination+1);
        if (des == 0) {
            if (type & PTYPE_TAG_DONTCOPY) {
                metanet_free(msg);
            }
            return -1;
        }
    } else {
        _filter_args(ctx, type, &session, (void**)&msg, &sz);
    }
    return metanet_send(ctx, source, des, type, session, msg, sz);
}

void 
metanet_callback(struct metanet_context *ctx, void *ud, metanet_cb cb) {
    ctx->cb = cb;
    ctx->cb_ud = ud; //模块部分的功能实体参数指针 比如logger的(struct logger*)
}

uint32_t 
metanet_current_handle(void) {
    if (G_NODE.init) {
        void *handle = pthread_getspecific(G_NODE.handle_key);
        return (uint32_t)(uintptr_t)handle;
    } else {
        uint32_t v = (uint32_t)(-THREAD_MAIN);
        return v;
    }
}

uint64_t
metanet_now(void) {
    return 0;
}